/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2022/5/22.
//

#include "sql/stmt/filter_stmt.h"
#include "common/lang/string.h"
#include "common/log/log.h"
#include "common/rc.h"
#include "sql/expr/expression.h"
#include "sql/parser/parse_defs.h"
#include "sql/stmt/select_stmt.h"
#include "storage/db/db.h"
#include "storage/field/field.h"
#include "storage/table/table.h"
#include <memory>
#include <string>
#include <unordered_map>

FilterStmt::~FilterStmt()
{
  for (FilterUnit *unit : filter_units_) {
    delete unit;
  }
  filter_units_.clear();
}

RC FilterStmt::create(Db *db, Table *default_table, std::unordered_map<std::string, Table *> *tables,
    ConditionSqlNode *conditions, int condition_num, FilterStmt *&stmt, unordered_map<string, string> *alias_map,
    unordered_map<string, string> *outside_alias_map)
{
  RC rc = RC::SUCCESS;
  stmt  = nullptr;

  FilterStmt *tmp_stmt = new FilterStmt();
  for (int i = 0; i < condition_num; i++) {
    FilterUnit *filter_unit = nullptr;

    rc = create_filter_unit(db, default_table, tables, conditions[i], filter_unit, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      delete tmp_stmt;
      LOG_WARN("failed to create filter unit. condition index=%d", i);
      return rc;
    }
    filter_unit->set_conjunction_type(conditions[i].conjunction_type);
    tmp_stmt->filter_units_.push_back(filter_unit);
  }

  stmt = tmp_stmt;
  return rc;
}

RC get_table_and_field(Db *db, Table *default_table, std::unordered_map<std::string, Table *> *tables,
    const RelAttrSqlNode &attr, Table *&table, const FieldMeta *&field)
{
  if (common::is_blank(attr.relation_name.c_str())) {
    table = default_table;
  } else if (nullptr != tables) {
    auto iter = tables->find(attr.relation_name);
    if (iter != tables->end()) {
      table = iter->second;
    }
  } else {
    table = db->find_table(attr.relation_name.c_str());
  }
  if (nullptr == table) {
    LOG_WARN("No such table: attr.relation_name: %s", attr.relation_name.c_str());
    return RC::SCHEMA_TABLE_NOT_EXIST;
  }

  field = table->table_meta().field(attr.attribute_name.c_str());
  if (nullptr == field) {
    LOG_WARN("no such field in table: table %s, field %s", table->name(), attr.attribute_name.c_str());
    table = nullptr;
    return RC::SCHEMA_FIELD_NOT_EXIST;
  }

  return RC::SUCCESS;
}

RC bind_field_in_arithmetic_expr(std::unique_ptr<Expression> &expr, Db *db, Table *default_table,
    std::unordered_map<std::string, Table *> *tables, unordered_map<string, string> *alias_map = nullptr,
    unordered_map<string, string> *outside_alias_map = nullptr)
{
  //
  if (expr->type() == ExprType::UNBOUND_AGGREGATION) {
    // 这里设计有点问题，实际上还是binder里边的内容
    auto                unbound_aggregate_expr = static_cast<UnboundAggregateExpr *>(expr.get());
    const char         *aggregate_name         = unbound_aggregate_expr->aggregate_name();
    AggregateExpr::Type aggregate_type;
    auto                f = [](const char *type_str, AggregateExpr::Type &type) -> RC {
      RC rc = RC::SUCCESS;
      if (0 == strcasecmp(type_str, "count")) {
        type = AggregateExpr::Type::COUNT;
      } else if (0 == strcasecmp(type_str, "sum")) {
        type = AggregateExpr::Type::SUM;
      } else if (0 == strcasecmp(type_str, "avg")) {
        type = AggregateExpr::Type::AVG;
      } else if (0 == strcasecmp(type_str, "max")) {
        type = AggregateExpr::Type::MAX;
      } else if (0 == strcasecmp(type_str, "min")) {
        type = AggregateExpr::Type::MIN;
      } else {
        rc = RC::INVALID_ARGUMENT;
      }
      return rc;
    };
    RC rc = f(aggregate_name, aggregate_type);
    if (OB_FAIL(rc)) {
      LOG_WARN("invalid aggregate name: %s", aggregate_name);
      return rc;
    }
    std::unique_ptr<Expression> child_expr = nullptr;
    if (unbound_aggregate_expr->child() != nullptr) {
      child_expr = std::move(unbound_aggregate_expr->child());
      rc         = bind_field_in_arithmetic_expr(child_expr, db, default_table, tables, alias_map, outside_alias_map);
      if (rc != RC::SUCCESS) {
        return rc;
      }
    }
    std::unique_ptr<Expression> aggregate_expr(
        static_cast<Expression *>(new AggregateExpr(aggregate_type, std::move(child_expr))));
    aggregate_expr->set_name(unbound_aggregate_expr->name());
    expr = std::move(aggregate_expr);
    return RC::SUCCESS;
  }
  // 这里其实还有可能是functionexpr,但是为了复用代码，这里不再另起一个函数了，直接使用这个函数
  if (expr->type() == ExprType::LENGTH) {
    LengthExpr *length_expr = static_cast<LengthExpr *>(expr.get());
    RC          rc =
        bind_field_in_arithmetic_expr(length_expr->child(), db, default_table, tables, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      return rc;
    }
  }
  if (expr->type() == ExprType::ROUND) {
    RoundExpr *round_expr = static_cast<RoundExpr *>(expr.get());
    RC rc = bind_field_in_arithmetic_expr(round_expr->child(), db, default_table, tables, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      return rc;
    }
  }
  if (expr->type() == ExprType::DATE_FORMAT) {
    DateFormatExpr *date_format_expr = static_cast<DateFormatExpr *>(expr.get());
    RC              rc               = bind_field_in_arithmetic_expr(
        date_format_expr->child(), db, default_table, tables, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      return rc;
    }
  }
  //  这里要递归调用，因为ArithmeticExpr里面可能还有ArithmeticExpr
  if (expr->type() != ExprType::ARITHMETIC && expr->type() != ExprType::UNBOUND_FIELD) {
    // 这里表示直接是一个Value,那么就直接返回
    return RC::SUCCESS;
  }
  if (expr->type() == ExprType::UNBOUND_FIELD) {
    // 这里是UnboundFieldExpr的情况
    // 这里需要把UnboundFieldExpr转换为FieldExpr
    UnboundFieldExpr *unbound_field_expr = static_cast<UnboundFieldExpr *>(expr.get());

    Table           *table = nullptr;
    const FieldMeta *field = nullptr;

    std::string table_name;
    bool        found_in_outside = false;
    if (alias_map == nullptr) {
      LOG_TRACE("alias_map is null");
      table_name = unbound_field_expr->table_name();
    } else {
      LOG_TRACE("alias_map is not null");
      auto iter = alias_map->find(unbound_field_expr->table_name());
      if (iter == alias_map->end()) {
        if (outside_alias_map != nullptr) {
          auto iter2 = outside_alias_map->find(unbound_field_expr->table_name());
          if (iter2 != outside_alias_map->end()) {
            table_name       = iter2->second;
            found_in_outside = true;
          } else {
            table_name = unbound_field_expr->table_name();
          }
        } else {
          table_name = unbound_field_expr->table_name();
        }
      } else {
        table_name = iter->second;
      }
    }
    RelAttrSqlNode attr = {table_name, unbound_field_expr->field_name()};
    RC             rc   = get_table_and_field(db, default_table, tables, attr, table, field);
    if (rc != RC::SUCCESS) {
      LOG_WARN("cannot find attr");
      return rc;
    }
    // 这里还要把FieldExpr转换为Expression指针
    std::unique_ptr<Expression> field_expr(static_cast<Expression *>(new FieldExpr(Field(table, field))));
    // 替换掉原来的expr
    static_cast<FieldExpr *>(field_expr.get())->set_correlated(found_in_outside);
    expr = std::move(field_expr);
    return RC::SUCCESS;
  }
  // 这里是ArithmeticExpr的情况
  ArithmeticExpr *arithmetic_expr = static_cast<ArithmeticExpr *>(expr.get());
  if (arithmetic_expr->left() != nullptr) {
    RC rc =
        bind_field_in_arithmetic_expr(arithmetic_expr->left(), db, default_table, tables, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      return rc;
    }
  }
  if (arithmetic_expr->right() != nullptr) {
    RC rc = bind_field_in_arithmetic_expr(
        arithmetic_expr->right(), db, default_table, tables, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      return rc;
    }
  }
  return RC::SUCCESS;
}

RC FilterStmt::create_filter_unit(Db *db, Table *default_table, std::unordered_map<std::string, Table *> *tables,
    ConditionSqlNode &condition, FilterUnit *&filter_unit, unordered_map<string, string> *alias_map,
    unordered_map<string, string> *outside_alias_map)
{
  RC                            rc = RC::SUCCESS;
  unordered_map<string, string> next_alias_map;
  if (alias_map != nullptr) {
    next_alias_map = *alias_map;
  }
  if (outside_alias_map != nullptr) {
    for (auto pair : *outside_alias_map) {
      next_alias_map.insert(pair);
    }
  }

  CompOp comp = condition.comp;
  if (comp < EQUAL_TO || comp >= NO_OP) {
    LOG_WARN("invalid compare operator : %d", comp);
    return RC::INVALID_ARGUMENT;
  }

  filter_unit = new FilterUnit;

  if (condition.comp == CompOp::EXISTS_OP || condition.comp == CompOp::NOT_EXISTS_OP) {
    // 这里是exists的情况，没有左值，只有右边一个子查询
    // 这里对子查询先从node解析成stmt，然后再存入其中
    SubQueryExpr *simple_sub_query = static_cast<SubQueryExpr *>(condition.right_expr.get());
    Stmt         *stmt             = nullptr;
    RC            rc1 = SelectStmt::create(db, *simple_sub_query->get_sql_node().get(), stmt, &next_alias_map);
    if (rc1 != RC::SUCCESS) {
      LOG_WARN("failed to create select stmt");
      return rc1;
    }
    if (stmt->type() != StmtType::SELECT) {
      LOG_WARN("wrong stmt type");
      return RC::INVALID_ARGUMENT;
    }

    // 把stmt存入expr中
    SelectStmt *select_stmt = static_cast<SelectStmt *>(stmt);
    simple_sub_query->set_select_stmt(select_stmt);

    // 最后设置右边的过滤单元
    FilterObj sub_query_filter_obj;
    sub_query_filter_obj.init_sub_query(std::move(condition.right_expr));
    filter_unit->set_right(std::move(sub_query_filter_obj));
    filter_unit->set_comp(condition.comp);
    return rc;
  }

  if (condition.right_expr != nullptr && condition.right_expr->type() == ExprType::TUPLE_VALUES) {
    // 这时候左边是一个attr，右边是一个直接的Tuple
    // 先设置左边的属性

    // 这里的left_expr可能是一个UnboundFieldExpr，也可能是ArithmeticExpr
    bind_field_in_arithmetic_expr(condition.left_expr, db, default_table, tables, alias_map, outside_alias_map);
    FilterObj filterObj;
    filterObj.init_obj_expr(std::move(condition.left_expr));
    filter_unit->set_left(std::move(filterObj));

    // 最后设置右边的过滤单元
    FilterObj tuple_values_filter_obj;
    tuple_values_filter_obj.init_tuple_values(std::move(condition.right_expr));
    filter_unit->set_right(std::move(tuple_values_filter_obj));
    filter_unit->set_comp(comp);
    return rc;
  }

  if (condition.left_expr != nullptr && condition.right_expr != nullptr &&
      condition.right_expr->type() == ExprType::SUB_QUERY &&
      (condition.left_expr->type() == ExprType::UNBOUND_FIELD || condition.left_expr->type() == ExprType::ARITHMETIC)) {
    // 这样的话左边就是属性，右边就是子查询
    // 先设置左边的属性

    // 这里的left_expr可能是一个UnboundFieldExpr，也可能是ArithmeticExpr
    bind_field_in_arithmetic_expr(condition.left_expr, db, default_table, tables, alias_map, outside_alias_map);
    FilterObj filterObj;
    filterObj.init_obj_expr(std::move(condition.left_expr));
    filter_unit->set_left(std::move(filterObj));

    SubQueryExpr *simple_sub_query = static_cast<SubQueryExpr *>(condition.right_expr.get());
    // 这里需要注意，需要查看一下表有多少列
    SelectSqlNode *sql_node = simple_sub_query->get_sql_node().get();
    if (sql_node->expressions.size() != 1) {
      LOG_WARN("sub query has more than one column");
      delete filter_unit;
      return RC::INVALID_ARGUMENT;
    }
    if (sql_node->expressions[0]->type() == ExprType::STAR) {
      // 还需要判断这个表一共有几个列
      const char *table_name = sql_node->relations_and_aliases[0].first.c_str();
      if (nullptr == table_name) {
        LOG_WARN("invalid argument. relation name is null.");
        delete filter_unit;
        return RC::INVALID_ARGUMENT;
      }

      Table *table = db->find_table(table_name);
      if (nullptr == table) {
        LOG_WARN("no such table. db=%s, table_name=%s", db->name(), table_name);
        delete filter_unit;
        return RC::SCHEMA_TABLE_NOT_EXIST;
      }
      TableMeta table_meta = table->table_meta();
      if (table_meta.field_num() - table_meta.sys_field_num() > 1) {
        LOG_WARN("sub query has more than one column");
        delete filter_unit;
        return RC::INVALID_ARGUMENT;
      }
    }

    // 这里对子查询先从node解析成stmt，然后再存入其中
    Stmt *stmt = nullptr;
    RC    rc1  = SelectStmt::create(db, *simple_sub_query->get_sql_node().get(), stmt, &next_alias_map);
    if (rc1 != RC::SUCCESS) {
      LOG_WARN("failed to create select stmt");
      return rc1;
    }
    if (stmt->type() != StmtType::SELECT) {
      LOG_WARN("wrong stmt type");
      return RC::INVALID_ARGUMENT;
    }

    // 把stmt存入expr中
    SelectStmt *select_stmt = static_cast<SelectStmt *>(stmt);
    simple_sub_query->set_select_stmt(select_stmt);

    // 最后设置右边的过滤单元
    FilterObj sub_query_filter_obj;
    sub_query_filter_obj.init_sub_query(std::move(condition.right_expr));
    filter_unit->set_right(std::move(sub_query_filter_obj));
    filter_unit->set_comp(comp);
    return rc;
  }

  else if (condition.left_expr != nullptr && condition.right_expr != nullptr &&
           condition.left_expr->type() == ExprType::SUB_QUERY &&
           (condition.right_expr->type() == ExprType::UNBOUND_FIELD ||
               condition.right_expr->type() == ExprType::ARITHMETIC)) {
    // 这样的话右边就是属性，左边就是子查询
    // 先设置右边的属性
    // 这里的right_expr可能是一个UnboundFieldExpr，也可能是ArithmeticExpr

    bind_field_in_arithmetic_expr(condition.right_expr, db, default_table, tables, alias_map, outside_alias_map);
    FilterObj filterObj;
    filterObj.init_obj_expr(std::move(condition.right_expr));
    filter_unit->set_right(std::move(filterObj));

    // 这里对子查询先从node解析成stmt，然后再存入其中
    SubQueryExpr *simple_sub_query = static_cast<SubQueryExpr *>(condition.left_expr.get());
    Stmt         *stmt             = nullptr;
    RC            rc1 = SelectStmt::create(db, *simple_sub_query->get_sql_node().get(), stmt, &next_alias_map);
    if (rc1 != RC::SUCCESS) {
      LOG_WARN("failed to create select stmt");
      return rc1;
    }
    if (stmt->type() != StmtType::SELECT) {
      LOG_WARN("wrong stmt type");
      return RC::INVALID_ARGUMENT;
    }

    // 把stmt存入expr中
    SelectStmt *select_stmt = static_cast<SelectStmt *>(stmt);
    simple_sub_query->set_select_stmt(select_stmt);

    // 最后设置右边的过滤单元
    FilterObj sub_query_filter_obj;
    sub_query_filter_obj.init_sub_query(std::move(condition.left_expr));
    filter_unit->set_left(std::move(sub_query_filter_obj));
    filter_unit->set_comp(comp);
    return rc;
  } else if (condition.left_expr != nullptr && condition.right_expr != nullptr &&
             condition.left_expr->type() == ExprType::SUB_QUERY &&
             condition.right_expr->type() == ExprType::SUB_QUERY) {
    // 这样的话左右两边都是子查询
    // 这里对子查询先从node解析成stmt，然后再存入其中
    SubQueryExpr *left_simple_sub_query = static_cast<SubQueryExpr *>(condition.left_expr.get());
    Stmt         *left_stmt             = nullptr;
    RC rc1 = SelectStmt::create(db, *left_simple_sub_query->get_sql_node().get(), left_stmt, &next_alias_map);
    if (rc1 != RC::SUCCESS) {
      LOG_WARN("failed to create select stmt");
      return rc1;
    }
    if (left_stmt->type() != StmtType::SELECT) {
      LOG_WARN("wrong stmt type");
      return RC::INVALID_ARGUMENT;
    }
    // 把stmt存入expr中
    SelectStmt *left_select_stmt = static_cast<SelectStmt *>(left_stmt);
    left_simple_sub_query->set_select_stmt(left_select_stmt);
    // 最后设置左边的过滤单元
    FilterObj left_sub_query_filter_obj;
    left_sub_query_filter_obj.init_sub_query(std::move(condition.left_expr));
    filter_unit->set_left(std::move(left_sub_query_filter_obj));

    // 右边的子查询
    SubQueryExpr *right_simple_sub_query = static_cast<SubQueryExpr *>(condition.right_expr.get());
    Stmt         *right_stmt             = nullptr;
    rc1 = SelectStmt::create(db, *right_simple_sub_query->get_sql_node().get(), right_stmt, &next_alias_map);
    if (rc1 != RC::SUCCESS) {
      LOG_WARN("failed to create select stmt");
      return rc1;
    }
    if (right_stmt->type() != StmtType::SELECT) {
      LOG_WARN("wrong stmt type");
      return RC::INVALID_ARGUMENT;
    }
    // 把stmt存入expr中
    SelectStmt *right_select_stmt = static_cast<SelectStmt *>(right_stmt);
    right_simple_sub_query->set_select_stmt(right_select_stmt);
    // 最后设置右边的过滤单元
    FilterObj right_sub_query_filter_obj;
    right_sub_query_filter_obj.init_sub_query(std::move(condition.right_expr));
    filter_unit->set_right(std::move(right_sub_query_filter_obj));
    filter_unit->set_comp(comp);
    return rc;
  } else {
    // 不属于特殊情况
    // 这里的left_expr和right_expr可能是UnboundFieldExpr，也可能是ArithmeticExpr
    RC rc = bind_field_in_arithmetic_expr(condition.left_expr, db, default_table, tables, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to bind field in arithmetic expr");
      return rc;
    }
    rc = bind_field_in_arithmetic_expr(condition.right_expr, db, default_table, tables, alias_map, outside_alias_map);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to bind field in arithmetic expr");
      return rc;
    }
    FilterObj left_filter_obj;
    left_filter_obj.init_obj_expr(std::move(condition.left_expr));
    filter_unit->set_left(std::move(left_filter_obj));
    FilterObj right_filter_obj;
    right_filter_obj.init_obj_expr(std::move(condition.right_expr));
    filter_unit->set_right(std::move(right_filter_obj));
  }

  // if (condition.left_is_attr) {
  //   Table           *table = nullptr;
  //   const FieldMeta *field = nullptr;
  //   rc                     = get_table_and_field(db, default_table, tables, condition.left_attr, table, field);
  //   if (rc != RC::SUCCESS) {
  //     LOG_WARN("cannot find attr");
  //     return rc;
  //   }
  //   FilterObj filter_obj;
  //   filter_obj.init_attr(Field(table, field));
  //   filter_unit->set_left(std::move(filter_obj));
  // } else {
  //   FilterObj filter_obj;
  //   filter_obj.init_value(condition.left_value);
  //   filter_unit->set_left(std::move(filter_obj));
  // }

  // if (condition.right_is_attr) {
  //   Table           *table = nullptr;
  //   const FieldMeta *field = nullptr;
  //   rc                     = get_table_and_field(db, default_table, tables, condition.right_attr, table, field);
  //   if (rc != RC::SUCCESS) {
  //     LOG_WARN("cannot find attr");
  //     return rc;
  //   }
  //   FilterObj filter_obj;
  //   filter_obj.init_attr(Field(table, field));
  //   filter_unit->set_right(std::move(filter_obj));
  // } else {
  //   FilterObj filter_obj;
  //   filter_obj.init_value(condition.right_value);
  //   filter_unit->set_right(std::move(filter_obj));
  // }

  filter_unit->set_comp(comp);

  // 检查两个类型是否能够比较
  return rc;
}
